package com.atguigu.flink.chapter05.sink;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @author liuyun
 * @title: Flink04_Sink_Mysql
 * @projectName flink0924
 * @description: TODO
 * @date 2022-3-6 19:22
 */
public class Flink04_Sink_Mysql {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> stream = env.socketTextStream("hadoop162", 9999)
                .map(line -> {
                    String[] data = line.split(",");
                    return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
                });

        stream.keyBy(WaterSensor::getId)
                .sum("vc")
                .addSink(new MysqlSink());

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    public static class MysqlSink extends RichSinkFunction<WaterSensor> {

        private Connection conn;

        @Override
        public void open(Configuration parameters) throws Exception {

            Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection("jdbc:mysql://hadoop162:3306/test", "root", "aaaaaa"); //alt+ctrl+enter


        }

        @Override
        public void close() throws Exception {
            if (conn != null && !conn.isClosed()) {
                conn.close();
            }
        }


        @Override
        public void invoke(WaterSensor value, Context context) throws Exception {

            String sql = "insert into sensor(id, ts, vc)values(?,?,?) on duplicate key update vc=?";
            PreparedStatement ps = conn.prepareStatement(sql);

            ps.setString(1, value.getId());
            ps.setLong(2, value.getTs());
            ps.setInt(3, value.getVc());
            ps.setInt(4, value.getVc());

            ps.execute();
            ps.close();
        }
    }
}
